package com.ykfa.juc;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.stream.Collectors;

/**
 *
 * 参考：https://zhuanlan.zhihu.com/p/360172630
 * @author yingkf
 * @date 2023年10月17日09:51:41
 */
public class CompletableFutureTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        runAsync();
       supplyAsync();
//        thenApply();
//        thenApplyAsync();
//        thenAccept();
//        thenRun();
//        thenCombine();
//        thenAcceptBoth();
//        runAfterBoth();
//        applyToEither();
//        thenCompose();//get(),join()
//        handle();
//        whenComplete();
//        allOf();
//        anyOf();
    }

    /**
     * 1. future接口
     * （1）isDone()
     * 判断任务是否完成。三种完成情况：normally（正常执行完毕）、exceptionally（执行异常）、via cancellation（取消）
     *
     * （2）get()
     * 阻塞获取结果或抛出受检测异常，需要显示进行try...catch处理。
     *
     * （3）get(long timeout,TimeUnit unit)
     * 超时阻塞获取结果
     *
     * （4）cancel(boolean mayInterruptIfRunning)
     * 取消任务，若一个任务未完成，则以CancellationException异常。其相关未完成的子任务也会以CompletionException结束
     *
     * （5）isCancelled()
     * 是否已取消，在任务正常执行完成前取消，才为true。否则为false。
     *
     * 2. CompletableFuture接口
     * （1）join
     * ​ 阻塞获取结果或抛出非受检异常。
     *
     * （2）getNow(T valueIfAbsent)
     * ​ 若当前任务无结果，则返回valueIfAbsent，否则返回已完成任务的结果。
     *
     * （3）complete(T value)
     * ​ 设置任务结果，任务正常结束，之后的任务状态为已完成。
     *
     * （4）completeExceptionally(Throwable ex)
     * ​ 设置任务异常结果，任务异常结束，之后的任务状态为已完成。
     *
     * （5）isCompletedExceptionally()
     * ​ 判断任务是否异常结束。异常可能的原因有：取消、显示设置任务异常结果、任务动作执行异常等。
     *
     * （6）getNumberOfDependents()
     * ​ 返回依赖当前任务的任务数量，主要用于监控。
     *
     * （7）orTimeout(long timeout,TimeUnit unit) jdk9
     * ​ 设置任务完成超时时间，若在指定时间内未正常完成，则任务会以异常(TimeoutException)结束。
     *
     * （8）completeOnTimeout(T value,long timeout,TimeUnit unit) jdk9
     * ​ 设置任务完成超时时间，若在指定时间内未正常完成，则以给定的value为任务结果
     */

    /**
     * runAsync用法，runAsync没有返回值
     */
    public static void runAsync() {
        //使用默认线程池
        CompletableFuture cf = CompletableFuture.runAsync(() -> System.out.println("Hello World!"));
        assert (cf.isDone());
        //使用自定义线程池
        CompletableFuture.runAsync(() -> System.out.println("Hello World!"), Executors.newSingleThreadExecutor());
    }

    /**
     * supplyAsync用法，有返回值
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void supplyAsync() throws ExecutionException, InterruptedException {
        CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {
            try {
                //ForkJoinPool.commonPool-worker-1线程
                System.out.println(Thread.currentThread().getName());
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        });
        //阻塞等待3秒
        String result = f.get();
        //main线程
        System.out.println(Thread.currentThread().getName());
        System.out.println(result);
    }

    /**
     * thenApplyAsync 跟thenApply区别就是thenApply是主线程执行的，thenApplyAsync是异步的，通常ForkJoinPool.commonPool()执行（通常，也可能不是），也可以自己指定线程池
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void thenApplyAsync() throws ExecutionException, InterruptedException {
        CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {
            System.out.println(s);
            return s.toUpperCase();
        }).thenApplyAsync(s -> {
            System.out.println(s);
            return s + "1";
        }).thenApplyAsync(s -> {
            System.out.println(s);
            return s + "2";
        })
//                .thenApplyAsync(s -> {
//            System.out.println(s);
//            return s + "3";
//        }, Executors.newSingleThreadExecutor())
                ;
        System.out.println(cf.get());
    }

    /**
     * thenApply 等上一阶段执行完成，使用上一阶段的结果，有返回值
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void thenApply() throws ExecutionException, InterruptedException {
        CompletableFuture cf = CompletableFuture.completedFuture("message");

        CompletableFuture c1 = cf.thenApply(s -> {
            System.out.println(s);
            return s + ":body";
        });
        Thread.sleep(10000);
        c1.thenApply(s -> {
            System.out.println(s);
            return s + ":body2";
        }).thenApply(s -> {
            System.out.println(s);
            return s + ":body3";
        }).thenApply(s -> {
            System.out.println(s);
            return s + ":body4";
        });
        System.out.println("执行了");
        System.out.println(cf.get());
        System.out.println("执行完了");
    }

    /**
     * 合并另一个CompletableFuture的结果，2个CompletableFuture都执行完了，再执行combine的逻辑，入参为2个任务结果，有返回值，combine是同步的
     */
    public static void thenCombine() {
        CompletableFuture<String> cfA = CompletableFuture.supplyAsync(() -> {
            System.out.println("processing a...");
            return "hello";
        });
        CompletableFuture<String> cfB = CompletableFuture.supplyAsync(() -> {
            System.out.println("processing b...");
            return " world";
        });
        CompletableFuture<String> cfC = CompletableFuture.supplyAsync(() -> {
            System.out.println("processing c...");
            return ", I'm CodingTao!";
        });
        cfA.thenCombine(cfB, (resultA, resultB) -> {
            System.out.println(resultA + resultB);  // hello world
            return resultA + resultB;
        }).thenCombine(cfC, (resultAB, resultC) -> {
            System.out.println(resultAB + resultC); // hello world, I'm CodingTao!
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return resultAB + resultC;
        });
        System.out.println("完成");
    }

    /**
     * 合并另外一个任务，两个任务都完成后，执行这个方法，入参为两个任务结果，不返回新结果
     */
    private static void thenAcceptBoth() {
        CompletableFuture<String> cfA = CompletableFuture.supplyAsync(() -> "resultA");
        CompletableFuture<String> cfB = CompletableFuture.supplyAsync(() -> "resultB");
        cfA.thenAcceptBoth(cfB, (resultA, resultB) -> {
            //resultA,resultB
            System.out.println(resultA + "," + resultB);
        });
    }

    /**
     * 无返回值无入参
     * <p>
     * 合并另外一个任务，两个任务都完成后，执行Runnable，注意，这里的两个任务是同时执行
     */
    private static void runAfterBoth() {
        CompletableFuture<String> cfA = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("process a");
            return "resultA";
        });
        CompletableFuture<String> cfB = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("process b");
            return "resultB";
        });
        cfB.runAfterBoth(cfA, () -> {
            //resultA,resultB
            System.out.println("任务A和任务B同时完成");
        });
        try {
            Thread.sleep(6000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * applyToEither其中任一任务完成后，执行Function，结果转换，入参为已完成的任务结果。返回新结果，要求两个任务结果为同一类型
     * acceptEither其中任一任务完成后，执行Consumer，消费结果，入参为已完成的任务结果。不返回新结果，要求两个任务结果为同一类型
     * runAfterEither其中任一任务完成后，执行Runnable，消费结果，无入参。不返回新结果，不要求两个任务结果为同一类型
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    private static void applyToEither() throws ExecutionException, InterruptedException {
        CompletableFuture<String> futureA = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "通过方式A获取商品a";
        });
        CompletableFuture<String> futureB = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "通过方式B获取商品a";
        });
        CompletableFuture<String> futureC = futureA.applyToEither(futureB, product -> "结果:" + product);
        //结果:通过方式A获取商品a
        System.out.println(futureC.get());
    }

    /**
     * thenCompose的用法
     * thenCompose 和thenApply的区别：
     * 两者差不多，但是thenCompose更简洁，apply返回的是一个future的future,thenCompose返回的是future
     * join 和 get 的区别 同样都是阻塞主线程获取任务执行结果，get可以抛出经检查的异常，可被捕获自定义处理或直接抛出，join会抛出未经检查的异常
     */
    private static void thenCompose() throws ExecutionException, InterruptedException {
        String original = "Message";
        CompletableFuture cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s))
                .thenCompose(upper -> CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s))
                        .thenApply(s -> upper + s));
        // MESSAGEmessage
        System.out.println(cf.join());

        CompletableFuture cf1 = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s))
                .thenApply(upper -> CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s))
                        .thenApply(s -> upper + s)).get();
        // MESSAGEmessage
        System.out.println(cf1.get());
    }

    private static String delayedUpperCase(String s) {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return s.toUpperCase();
    }

    private static String delayedLowerCase(String s) {
        try {
            Thread.sleep(300);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return s.toLowerCase();
    }

    /**
     * handle的用法，任务完成后执行BiFunction，结果转换，入参为结果或者异常，返回新结果
     */
    private static void handle() {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "resultA")
                .thenApply(resultA -> resultA + " resultB")
                // 任务 C 抛出异常
                .thenApply(resultB -> {
                    throw new RuntimeException();
                })
                // 处理任务 C 的返回值或异常
                .handle(new BiFunction<Object, Throwable, Object>() {
                    @Override
                    public Object apply(Object re, Throwable throwable) {
                        if (throwable != null) {
                            return "errorResultC";
                        }
                        return re;
                    }
                })
                .thenApply(resultC -> {
                    System.out.println("resultC:" + resultC);
                    return resultC + " resultD";
                });
        System.out.println(future.join());
    }

    /**
     * whenComplete的用法，任务完成后执行BiConsumer，结果消费，入参为结果或者异常，不返回新结果,如果上一阶段执行有异常，这个会在whenComplete的结果.get的时候抛出异常
     * exceptionally任务异常，则执行Function，异常转换，入参为原任务的异常信息，若原任务无异常，则返回原任务结果，即不执行转换TODO
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    private static void whenComplete() throws ExecutionException, InterruptedException {
        // 创建异步执行任务:
        CompletableFuture<Double> cf = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + "job1 start,time->" + System.currentTimeMillis());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
            }
            if (true) {
                throw new RuntimeException("test");
            } else {
                System.out.println(Thread.currentThread() + "job1 exit,time->" + System.currentTimeMillis());
                return 1.2;
            }
        });
        //cf执行完成后会将执行结果和执行过程中抛出的异常传入回调方法
        // 如果是正常执行，a=1.2，b则传入的异常为null
        //如果异常执行，a=null，b则传入异常信息
        CompletableFuture<Double> cf2 = cf.whenComplete((a, b) -> {
            System.out.println(Thread.currentThread() + "job2 start,time->" + System.currentTimeMillis());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
            }
            if (b != null) {
                System.out.println("error stack trace->");
                b.printStackTrace();
            } else {
                System.out.println("run succ,result->" + a);
            }
            System.out.println(Thread.currentThread() + "job2 exit,time->" + System.currentTimeMillis());
        });
        //等待子任务执行完成
        System.out.println("main thread start wait,time->" + System.currentTimeMillis());
        //如果cf是正常执行的，cf2.get的结果就是cf执行的结果
        //如果cf是执行异常，则cf2.get会抛出异常
        System.out.println("run result->" + cf2.get());
        System.out.println("main thread exit,time->" + System.currentTimeMillis());
    }

    /**
     * thenAccept 用法，等上一阶段完成，结果传入，再执行，无返回结果
     *
     * @throws InterruptedException
     */
    public static void thenAccept() throws InterruptedException {
        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> "message").thenAccept(System.out::println);
    }

    /**
     * thenRun 用法，等上一阶段完成，再执行，无入参，无返回结果
     *
     * @throws InterruptedException
     */
    public static void thenRun() throws InterruptedException {
        CompletableFuture.supplyAsync(() -> {
            //执行异步任务
            System.out.println("执行任务");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "success";
        }).thenRun(() -> System.out.println("上一阶段任务执行完成"));
        Thread.sleep(2000);
    }

    /**
     * anyOf 全部CompletableFuture完成，才返回一个新的CompletableFuture，要求anyOf的入参的CompletableFuture的返回结果类型一致
     * 我们在处理业务时，有时会有多任务异步处理，同步返回结果的情况
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    private static void allOf() throws ExecutionException, InterruptedException {
        // 创建异步执行任务:
        CompletableFuture<Double> cf = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " start job1,time->" + System.currentTimeMillis());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread() + " exit job1,time->" + System.currentTimeMillis());
            return 1.2;
        });
        CompletableFuture<Double> cf2 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " start job2,time->" + System.currentTimeMillis());
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread() + " exit job2,time->" + System.currentTimeMillis());
            return 3.2;
        });
        CompletableFuture<Double> cf3 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " start job3,time->" + System.currentTimeMillis());
            try {
                Thread.sleep(1300);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
//            throw new RuntimeException("test");
            System.out.println(Thread.currentThread() + " exit job3,time->" + System.currentTimeMillis());
            return 2.2;
        });
        //allof等待所有任务执行完成才执行cf4，如果有一个任务异常终止，则cf4.get时会抛出异常，都是正常执行，cf4.get返回null
        //anyOf是只有一个任务执行完成，无论是正常执行或者执行异常，都会执行cf4，cf4.get的结果就是已执行完成的任务的执行结果
        CompletableFuture cf4 = CompletableFuture.allOf(cf, cf2, cf3).whenComplete((a, b) -> {
            if (b != null) {
                System.out.println("error stack trace->");
                b.printStackTrace();
            } else {
                System.out.println("run succ,result->" + a);
            }
        });
        CompletableFuture<List> rs = CompletableFuture.allOf(cf, cf2, cf3).thenApply(v -> Arrays.asList(cf, cf2, cf3).stream().map(future -> future.join())
                .collect(Collectors.toList()));
        System.out.println("main thread start cf4.get(),time->" + System.currentTimeMillis());
        //等待子任务执行完成
        System.out.println("cf4 run result->" + cf4.get());
        System.out.println("main thread exit,time->" + System.currentTimeMillis());
        rs.get().forEach(o -> System.out.println("cf执行结果：" + o));
    }

    /**
     * anyOf 任意一个CompletableFuture完成，即返回一个新的CompletableFuture，要求anyOf的入参的CompletableFuture的返回结果类型一致
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    private static void anyOf() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            return "Result of Future 1";
        });

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            return "Result of Future 2";
        });

        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            return "Result of Future 3";
        });
        CompletableFuture<Object> anyOfFuture = CompletableFuture.anyOf(future1, future2, future3);
        System.out.println(anyOfFuture.get()); // Result of Future 2
    }
}
